package pro.fraud_detect.sources;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import pro.fraud_detect.beans.Transaction;
import pro.fraud_detect.config.Config;
import pro.fraud_detect.config.Parameters;

import java.util.Properties;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/2/22
 **/
public class InputSource {

    public static SourceFunction<String> createRuleSource(StreamExecutionEnvironment env, String sourceType, Config config) {

        switch (sourceType) {
            case "kafka":
                Properties kafkaProps = new Properties();
                String kafkaHost = config.get(Parameters.KAFKA_HOST);
                int kafkaPort = config.get(Parameters.KAFKA_PORT);
                String servers = String.format("%s:%s", kafkaHost, kafkaPort);
                kafkaProps.setProperty("bootstrap.servers", servers);
                kafkaProps.setProperty("auto.offset.reset", config.get(Parameters.OFFSET));

                FlinkKafkaConsumer consumer = new FlinkKafkaConsumer(config.get(Parameters.RULES_TOPIC), new SimpleStringSchema(), kafkaProps);
                return consumer;
            case "socket":
                new SocketTextStreamFunction("localhost",config.get(Parameters.SOCKET_PORT),"\n",3);
            default:
                throw new IllegalArgumentException("source type "+sourceType+" is unknown");
        }
    }

    public static SourceFunction<String> createTransactionDataSource(StreamExecutionEnvironment env, String sourceType, Config config) {

        switch (sourceType) {
            case "kafka":
                Properties kafkaProps = new Properties();
                String kafkaHost = config.get(Parameters.KAFKA_HOST);
                int kafkaPort = config.get(Parameters.KAFKA_PORT);
                String servers = String.format("%s:%s", kafkaHost, kafkaPort);
                kafkaProps.setProperty("bootstrap.servers", servers);
                kafkaProps.setProperty("auto.offset.reset", config.get(Parameters.OFFSET));

                FlinkKafkaConsumer consumer = new FlinkKafkaConsumer(config.get(Parameters.DATA_TOPIC), new SimpleStringSchema(), kafkaProps);
                return consumer;
            case "socket":
                new SocketTextStreamFunction("localhost",config.get(Parameters.SOCKET_PORT),"\n",3);
            default:
                throw new IllegalArgumentException("source type "+sourceType+" is unknown");
        }

    }

    public static Transaction string2Transaction(String line){
        Transaction transaction = JSON.parseObject(line, Transaction.class);
        transaction.setIngestionTimestamp(System.currentTimeMillis());
        return transaction;
    }

}
